package me.sdau.spider.kafkaprocess.kafka.dsl;

import me.sdau.spider.kafkaprocess.comm.utils.FileUtils;
import me.sdau.spider.kafkaprocess.kafka.dto.ConfigDTO;
import me.sdau.spider.kafkaprocess.kafka.nlp.NLPBuilder;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * @author zj
 * @date 2018/3/29 17:21
 */
public class App {

    public static void WordCount() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-counts");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "172.16.18.6:9092");
//        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> stream = builder.stream("testI");

        /*stream.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
                .groupBy((k, v) -> v)
                .count(Materialized.as("counts"))
                .toStream()
                .to("testII", Produced.with(Serdes.String(), Serdes.Long()));*/

        /*stream.map((key, value) -> {
            System.out.println("key: " + key);
            System.out.println("value:" + value);
            JSONObject json = JSONObject.parseObject(value);
            json.put("type", "stream");
            return KeyValue.pair(key, json.toString());
        }).to("testII");*/

        /*stream.map((key, value) -> {
            System.out.println("val ==>" + value);
            return KeyValue.pair(key, value);
        }).to("testII");*/

        /*stream.map((key, value) -> {
            System.out.println("key==>" + key + "val<==" + value);
            JSONObject json = JSONObject.parseObject(value);
            String title = json.getString("title");
            String content = json.getString("content");
            String summary = new Summary().get(title, content);
            json.put("summary", summary);
            return KeyValue.pair(key, json.toString());
        });*/

        Topology topology = builder.build();
        KafkaStreams streams = new KafkaStreams(topology, props);
        streams.start();
    }

    public static List<KafkaStreams> process() {
        List<ConfigDTO> index = Optional.ofNullable(FileUtils.getIndex()).orElse(new ArrayList<>());
        System.out.println(index);
        return index
                .stream()
                .map(NLPBuilder::getStreams)
                .collect(Collectors.toList());
    }
}
